package com.alpaca.recommend.hbase.activerecord;

import com.alpaca.recommend.hbase.HTableClients;
import com.alpaca.recommend.util.SpringContextLoader;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterBase;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * Created by wangj on 2015/2/9.
 */
public abstract class HModel<M extends HModel> implements Serializable {

    /**
     * Attributes of this model
     */
    private Result result;

    private Put put;

    private Delete delete;

    private Table table;

    public boolean exists() {
        return result == null || result.getRow() == null ? false : true;
    }


    public M setRow(String row) {
        put = new Put(row.getBytes());
        return (M) this;
    }


    public M set(String family, String qualifier, String value) {
        return set(family, qualifier, Bytes.toBytes(value));
    }

    public M set(String family, String qualifier, Integer value) {
        return set(family, qualifier, Bytes.toBytes(value));
    }

    public M set(String family, String qualifier, Long value) {
        return set(family, qualifier, Bytes.toBytes(value));
    }

    public M set(String family, String qualifier, byte[] value) {
        if (put == null) {
            throw new RuntimeException("please setRow first!");
        }

        put.addColumn(family.getBytes(), qualifier.getBytes(), value);
        return (M) this;
    }

    public M get(String row) {
        return get(row.getBytes());
    }

    /**
     * 保存对象
     */
    public void save() {
        try {
            getTable().put(put);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void delete(String rowkey) {
        try {
            Delete delete = new Delete(rowkey.getBytes());
            getTable().delete(delete);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public boolean checkAndPut(String row, String family, String qualifier, String value) {
        try {
            return getTable().checkAndPut(Bytes.toBytes(row), Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value), put);
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    public boolean checkAndPut(String row, String family, String qualifier, Integer value) {
        try {
            return getTable().checkAndPut(Bytes.toBytes(row), Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value), put);
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }


    public String getRowKey() {
        return Bytes.toString(result.getRow());
    }

    public M get(byte[] row) {
        try {
            result = getTable().get(new Get(row));
            if (result != null && result.getRow() != null) {
                M object = build(result, getClass());
                return object;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 多个row key 查询
     *
     * @param rows
     * @return
     */
    public List<M> find(String... rows) {
        List<String> keys = new ArrayList<String>();
        Collections.addAll(keys, rows);
        return find(keys);
    }

    public List<M> find(List<String> rows) {
        List<M> models = new ArrayList<M>();

        List<Get> gets = new ArrayList<Get>(rows.size());
        for (String row : rows) {
            gets.add(new Get(row.getBytes()));
        }
        try {
            Table hTableInterface = getTable();
            Result[] results = hTableInterface.get(gets);

            for (Result result : results) {
                M object = build(result, getClass());
                models.add(object);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return models;
    }

    private Table getTable() throws IOException {
        if (table != null) {
            return table;
        }

        HTableClients hTableClients = SpringContextLoader.getBean(HTableClients.class);
        table = hTableClients.getHTable(TableMapping.me().getTable(getClass()));
        return table;
    }

    /**
     * 前缀查询
     *
     * @param rowPrefix
     * @return
     */
    public List<M> findByPrefix(String rowPrefix) {
        return findByFilter(new PrefixFilter(rowPrefix.getBytes()));
    }

    public List<M> findByFilter(FilterBase filterBase) {
        Scan scan = new Scan();
        scan.setFilter(filterBase);
        return findByScan(scan);
    }

    /**
     * 根据起始行查询
     *
     * @param startRow
     * @param endRow
     * @return
     */
    public List<M> find(String startRow, String endRow) {
        Scan s = new Scan();
        s.setStartRow(startRow.getBytes());
        s.setStopRow(endRow.getBytes());

        return findByScan(s);
    }

    public List<M> findByScan(Scan scan) {
        List<M> models = new ArrayList<M>();
        try {
            ResultScanner rs = getTable().getScanner(scan);
            for (Result r : rs) {
                M object = build(r, getClass());
                models.add(object);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return models;
    }

    public byte[] getValue(byte[] family, byte[] qualifier) {
        return result.getValue(family, qualifier);
    }

    public String getString(String family, String qualifier) {
        return Bytes.toString(getValue(family.getBytes(), qualifier.getBytes()));
    }

    public int getInt(String family, String qualifier) {
        return getValue(family.getBytes(), qualifier.getBytes()) == null ? 0 :
                Bytes.toInt(getValue(family.getBytes(), qualifier.getBytes()));
    }

    public long getLong(String family, String qualifier) {
        return getValue(family.getBytes(), qualifier.getBytes()) == null ? 0 :
                Bytes.toLong(getValue(family.getBytes(), qualifier.getBytes()));
    }

    public final M build(Result result, Class<? extends HModel> modelClass) {
        try {
            HModel<?> ar = modelClass.newInstance();
            ar.setResult(result);
            ar.setRow(Bytes.toString(result.getRow()));
            return (M) ar;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public void setResult(Result result) {
        this.result = result;
    }
}
